package worker

import (
	"context"
	"github.com/mongodb/mongo-go-driver/mongo"
	"github.com/mongodb/mongo-go-driver/mongo/clientopt"
	"go-crontab/common"
	"time"
)

// 保存日志模块
type LogSink struct {
	client         *mongo.Client
	logCollection  *mongo.Collection
	logChan        chan *common.JobLog
	autoCommitChan chan *common.LogBatch
}

var (
	G_logSink *LogSink
)

// 批量写入日志
func (logSink *LogSink) saveLogs(batch *common.LogBatch) {
	logSink.logCollection.InsertMany(context.TODO(), batch.Logs)
	// TODO: 保存日志是否成功？
}

func (logSink *LogSink) writeLoop() {
	var (
		log      *common.JobLog
		logBatch *common.LogBatch // 当前的日志批次
		// 让这个批次超时自动提交
		commitTimer  *time.Timer
		timeoutBatch *common.LogBatch // 超时批次
	)

	for {
		select {
		case log = <-logSink.logChan:
			if logBatch == nil {
				logBatch = &common.LogBatch{}
				commitTimer = time.AfterFunc(time.Duration(G_config.JobLogCommitTimeout)*time.Millisecond,
					// 因为这儿的func是另外一个协程，和writeLopp不是同一个协程。
					// scheduler会并发调度(两个协程操作了logBatch变量)，所以这儿只发出通知，不能提交batch
					func(logBatch *common.LogBatch) func() {
						return func() {
							logSink.autoCommitChan <- logBatch
						}
					}(logBatch)) // 因为外层的logBatch变量内存地址一直在变,所以用闭包始终引用变化后的那个地址
			}
			logBatch.Logs = append(logBatch.Logs, log)
			if len(logBatch.Logs) >= G_config.JobLogBatchSize {
				logSink.saveLogs(logBatch)
				logBatch = nil
				commitTimer.Stop() // 需要取消掉定时器
			}
		case timeoutBatch = <-logSink.autoCommitChan:
			// 判断过期批次仍旧是当前批次,防止重复提交日志批次
			if timeoutBatch == logBatch {
				logSink.saveLogs(timeoutBatch)
				logBatch = nil
			}
		}
	}
}

func InitLogSink() (err error) {
	var (
		client *mongo.Client
	)

	if client, err = mongo.Connect(
		context.TODO(),
		G_config.MongodbUri,
		clientopt.ConnectTimeout(time.Duration(G_config.MongodbConnectTimeout)*time.Second)); err != nil {
		return
	}

	G_logSink = &LogSink{
		client:         client,
		logCollection:  client.Database("cron").Collection("log"),
		logChan:        make(chan *common.JobLog, 1000),
		autoCommitChan: make(chan *common.LogBatch, 1000),
	}
	// 启动日志存储协程
	go G_logSink.writeLoop()

	return
}

func (logSink *LogSink) Append(jobLog *common.JobLog) {
	select {
	case logSink.logChan <- jobLog:
	default:
		// 队列满了就丢弃
	}
}
